package com.lagou.no6

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.api.scala._
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

import scala.util.Random
object TableDemo {
    def main(args: Array[String]): Unit = {
        //配置环境参数
        val settings = EnvironmentSettings.newInstance()
          .useBlinkPlanner()
          .inStreamingMode()
          .build();
        //获取流环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //创建表环境
        val tEnv = StreamTableEnvironment.create(env,settings)
        //根据socket生成kv的元组
        val dataDS: DataStream[(String, Int)] = env.socketTextStream("node01",9999)
          .map{line =>
              val arr = line.split(",")
              (arr(0).trim,arr(1).trim.toInt)
          }
        //根据流对象生成表，指定字段名
        tEnv.createTemporaryView("t",dataDS,$("name"),$("num"))
        //定义sql，求每个key的num和
        val sql = "select name,sum(num) as sum_num from t group by name"
        //执行sql，返回一个表对象
        val tableRes: Table = tEnv.sqlQuery(sql)
        //将表转为流
        val res: DataStream[(Boolean, Row)] = tEnv.toRetractStream[Row](tableRes)
        //输出
        res.print()
        env.execute()
    }
}
